package net.bwie.realtime.dws.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dws.log.bean.MonitorInfo;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import utils.KafkaUtil;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author ASUS
 */
public class DangerousDriverController {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 每 1000ms 开始一次 checkpoint
        // 5秒一次 Checkpoint
        env.enableCheckpointing(5000);

        // 获取 kafka 数据源
        DataStream<String> ds1 = KafkaUtil.consumerKafka(env, "traffic_monitor_data");

//        ds1.print();

        // 转换数据结构为实体类
        SingleOutputStreamOperator<MonitorInfo> ds2 = ds1.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                return new MonitorInfo(
                        jsonObject.getLong("actionTime"),
                        jsonObject.getString("car"),
                        jsonObject.getString("monitorId"),
                        jsonObject.getString("cameraId"),
                        jsonObject.getInteger("speed"),
                        jsonObject.getString("roadId"),
                        jsonObject.getString("areaId"),
                        jsonObject.getInteger("speedLimit"));
            }
        });



//        monitor.print();

        // 水位线  允许数据迟到五秒
        SingleOutputStreamOperator<MonitorInfo> ds3 = ds2.assignTimestampsAndWatermarks(WatermarkStrategy
                .<MonitorInfo>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<MonitorInfo>() {
                    @Override
                    public long extractTimestamp(MonitorInfo monitorInfo, long l) {
                        return monitorInfo.getActionTime();
                    }
                }));


//        ds3.print();

        // 定义Pattern
        Pattern<MonitorInfo, MonitorInfo> pattern = Pattern.<MonitorInfo>begin("first", AfterMatchSkipStrategy.skipPastLastEvent())
                .where(new SimpleCondition<MonitorInfo>() {
                    @Override
                    public boolean filter(MonitorInfo monitorInfo) throws Exception {
                        if (monitorInfo.getSpeedLimit() == null || monitorInfo.getSpeed() == null) {
                            return false;
                        }
                        int limitSpeed = monitorInfo.getSpeedLimit();
                        int speed = monitorInfo.getSpeed();
                        return speed > limitSpeed * 1.2;
                    }
                })
                // TODO 匹配3次满足条件的事件
                .times(3).allowCombinations()
                // TODO 满足2分钟内条件
                .within(Time.minutes(2));


        // CEP
        PatternStream<MonitorInfo> patternStream = CEP.pattern(ds3.keyBy(MonitorInfo::getCar), pattern);

        // 输出数据
        SingleOutputStreamOperator<String> ds4 = patternStream.select(new PatternSelectFunction<MonitorInfo, String>() {
            @Override
            public String select(Map<String, List<MonitorInfo>> map) throws Exception {
//                return map.get("first").get(0).getCar();
                List<MonitorInfo> dangerousList = map.get("dangerousEvent");
                if (dangerousList == null || dangerousList.size() < 3) {
                    return ""; // 兜底：避免异常数据
                }
                // 提取关键信息（车牌号、3次超速时间、路段、速度）
                MonitorInfo first = dangerousList.get(0);
                MonitorInfo second = dangerousList.get(1);
                MonitorInfo third = dangerousList.get(2);

                // 格式化输出为JSON（便于后续写入Kafka/Doris存储）
                JSONObject result = new JSONObject();
                result.put("car", first.getCar());
                result.put("firstSpeed", first.getSpeed());
                result.put("firstTime", first.getActionTime()); // 建议转成yyyy-MM-dd HH:mm:ss格式
                result.put("firstRoad", first.getRoadId());
                result.put("secondSpeed", second.getSpeed());
                result.put("secondTime", second.getActionTime());
                result.put("secondRoad", second.getRoadId());
                result.put("thirdSpeed", third.getSpeed());
                result.put("thirdTime", third.getActionTime());
                result.put("thirdRoad", third.getRoadId());
                result.put("riskLevel", "严重超速（2分钟内3次）"); // 增加风险等级，便于业务分类

                return result.toJSONString();
            }
        });

        ds4.print("涉嫌危险驾驶的车牌号是：");

        env.execute("DangerousDriverController");
    }
}

